package com.chujian.framework.ws.support;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.chujian.framework.constant.ThreadConst;
import com.chujian.framework.ws.*;
import com.chujian.framework.ws.support.rcv.IWsDbStoreService;
import com.chujian.framework.ws.support.rcv.RcvMsgRespDataCenter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.OrderComparator;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 缺省的发送器（也是发送调度）
 */
@Slf4j
public class DefaultWsSender implements IWsSender,
        InitializingBean, ApplicationContextAware {

    private final IWsSendExecutor sendExecutor;
    private final ThreadPoolTaskExecutor taskExecutor;
    private final IWsDbStoreService wsDbStoreService;
    private ApplicationContext ctx;

    /**
     * 数据接收中心
     */
    private final RcvMsgRespDataCenter msgRespDataCenter = new RcvMsgRespDataCenter();
    /**
     * 使用 LinkedBlockingDeque 做消息缓冲
     */
    private final LinkedBlockingDeque<WsMsg> msgQueue = new LinkedBlockingDeque<>();
    private final List<IWsSendMsgWrapper> sendMsgWrappers = new ArrayList<>();

    public DefaultWsSender(IWsSendExecutor sendExecutor,
                           @Qualifier(ThreadConst.CORE_EXECUTOR) ThreadPoolTaskExecutor taskExecutor,
                           @Nullable IWsDbStoreService wsDbStoreService) {
        this.sendExecutor = sendExecutor;
        this.taskExecutor = taskExecutor;
        this.wsDbStoreService = wsDbStoreService;

    }

    @Override
    public void send(WsMsg msg) throws Exception {
        msgQueue.add(msg);
    }

    @Override
    public WsResult sendDirect(WsMsg msg) throws Exception {
        return callSend(msg);
    }

    @Override
    public WsMsg sendAndResp(WsMsg msg) throws Exception {
        callSend(msg);
        return msgRespDataCenter.takeResp(msg.getReqId());
    }

    /**
     * 执行发送，返回发送成功标记
     *
     * @param msg
     * @return
     */
    private WsResult callSend(WsMsg msg) {
        // 同步前
        msg = beforeSend(msg);
        if (null == msg) {
            return new WsResult(WsResult.FAIL_CODE, "beforeSend阻止消息发送");
        }
        // 执行同步
        return doSend(msg);
    }

    private WsResult doSend(WsMsg msg) {
        log.info("发送消息：{}", msg.simpleToString());
        try {
            sendExecutor.send(msg);
            afterSendSuc(msg);
            return new WsResult(WsResult.SUC_CODE, "发送成功");
        } catch (Exception e) {
            String errMsg = "同步器消息发送失败: " + e.getMessage();
            afterSendFail(msg, e, errMsg);
            return new WsResult(WsResult.FAIL_CODE, errMsg);
        }
    }

    private WsMsg beforeSend(WsMsg msg) {
        // #. 数据包装
        for (IWsSendMsgWrapper wrapper : sendMsgWrappers) {
            msg = wrapper.wrapper(msg);
            if (null == msg) {
                return null;
            }
        }
        // #. 数据库保存
        if (null != wsDbStoreService) {
            wsDbStoreService.saveWsMsg(msg);
        }
        return msg;
    }

    private void afterSendSuc(WsMsg msg) {
        // 1. 标记发送成功
        if (null != wsDbStoreService) {
            wsDbStoreService.markSendSuccess(msg);
        }
    }

    private void afterSendFail(WsMsg msg, Exception e, String errMsg) {
        log.error("同步器消息发送失败", e);

        // 累加失败次数
        msg.addErrCnt();

        // 数据库记录失败信息
        if (null != wsDbStoreService) {
            wsDbStoreService.markSendFail(msg, errMsg);
        }
    }


    // 发送缓存的消息
    private void startSendCache() {
        taskExecutor.submit(() -> {
            WsMsg msg;
            while (true) {
                try {
                    msg = msgQueue.take();
                    callSend(msg);
                } catch (InterruptedException interruptedException) {
                    log.warn("ws发送线程退出");
                    break;
                } catch (Exception e) {
                    log.error("消息发送失败", e);
                }
                ThreadUtil.sleep(10);
            }
        });
    }

    /**
     * 重试，发送失败的消息
     */
    @Scheduled(fixedDelay = 15 * 1000)
    public void retrySend() {
        if (null == wsDbStoreService) {
            return;
        }
        List<WsMsg> msgs = wsDbStoreService.listRetrySendMsg(20);
        if (CollUtil.isNotEmpty(msgs)) {
            for (WsMsg msg : msgs) {
                doSend(msg);
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        startSendCache();
        collectMsgWrapper();
    }

    // 收集所有的消息包装器
    private void collectMsgWrapper() {
        Collection<IWsSendMsgWrapper> wrappers = ctx.getBeansOfType(IWsSendMsgWrapper.class).values();
        sendMsgWrappers.addAll(wrappers);
        sendMsgWrappers.sort(OrderComparator.INSTANCE);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.ctx = applicationContext;
    }
}
